/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.util;

import java.util.ArrayList;
import java.util.Iterator;

import org.singularity.io.DME;
import org.singularity.io.Pipe;
import org.singularity.io.DME.Code;
import org.singularity.mapred.RedFun;

/**
 * @author vjache
 *
 */
public abstract class AbstractCartesianRedFun<KT, RT> implements RedFun<KT, RT> {
	
	public static interface Accumulator<RT>
	{
		void onToken(DME.Code aDmeCode, Pipe<DME<RT>> aOut, Object[] aToken) throws Exception;
		void onFinish(DME.Code aDmeCode, Pipe<DME<RT>> aOut) throws Exception;
	}
	

	@Override
	public void red(
			DME.Code aDmeCode, 
			Pipe<DME<RT>> aOut, 
			KT aKey, 
			Iterable<?>... aSgments) throws Exception 
	{
		Object[] cursor = new Object[aSgments.length];
		final Accumulator<RT> acc = createAccumulator(aKey);
		if(traverse(aDmeCode,0,aOut,aSgments,cursor,acc))
			acc.onFinish(aDmeCode, aOut);
	};
	
	/**
	 * Creates new accumulator per reduce call.
	 * @return
	 */
	protected abstract Accumulator<RT> createAccumulator(KT aKey);
	
	private boolean traverse(
			DME.Code aDmeCode,
			int aPos,
			Pipe<DME<RT>> aOut,
			Iterable<?>[] aSgments, 
			Object[] cursor,
			Accumulator<RT> aAcc) throws Exception
	{
		if(aPos==aSgments.length)
		{
			aAcc.onToken(aDmeCode,aOut,cursor);
			return true;
		}
		final Iterator<?> it = aSgments[aPos].iterator();
		if(!it.hasNext())
			return false;
		while(it.hasNext()){
			final Object dme = it.next();
			cursor[aPos]=dme;
			if(!traverse(aDmeCode,aPos+1,aOut,aSgments,cursor,aAcc))
				return false;
		}
		return true;
	}
	
	public static void main(String[] args) throws Exception {
		ArrayList<String> names = new ArrayList<String>();
		names.add("Vasya");
		names.add("vjache");
		names.add("Petya");
		names.add("Kolya");
		names.add("Senya");
		ArrayList<String> families = new ArrayList<String>();
		families.add("Pupkin");
		families.add("Mumkin");
		families.add("Bumkin");
		families.add("Fumkin");
		families.add("Hrumkin");
		families.add("Bulba");
		ArrayList<Integer> ages = new ArrayList<Integer>();
		for(int i=1;i<1000;i++)
		{
			ages.add(i);
		}
		ArrayList<Integer> weights = new ArrayList<Integer>();
		for(int i=1;i<1000;i++)
		{
			weights.add(i);
		}
		
		
		final AbstractCartesianRedFun<Object,Object> cFun = new AbstractCartesianRedFun<Object,Object>() {

			@Override
			protected Accumulator<Object> createAccumulator(Object aKey) {
				return new Accumulator<Object>() {
					int cnt =0;
					@Override
					public void onFinish(Code aDmeCode, Pipe<DME<Object>> aOut) 
					{System.out.println(cnt);}

					@Override
					public void onToken(Code aDmeCode, Pipe<DME<Object>> aOut,
							Object[] aToken) {cnt++;}
				};
			}

			
		};
		cFun.red(DME.Code.Add,null, null, names,families,ages,weights);
	}

}
